/*
 * Copyright 2018 datagear.tech
 *
 * Licensed under the LGPLv3 license:
 * http://www.gnu.org/licenses/lgpl-3.0.html
 */

package org.datagear.management.domain;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.datagear.analysis.DataSetException;
import org.datagear.analysis.DataSetProperty;
import org.datagear.analysis.DataSetProperty.DataType;
import org.datagear.analysis.DataSetQuery;
import org.datagear.analysis.ResolvedDataSetResult;
import org.datagear.analysis.support.DataSetPropertyValueConverter;
import org.datagear.analysis.support.SqlDataSet;
import org.datagear.analysis.support.SqlDataSetUnsupportedSqlTypeException;
import org.datagear.analysis.support.TemplateResolvedDataSetResult;
import org.datagear.management.util.BigQueryClient;
import org.datagear.util.resource.ConnectionFactory;
import org.springframework.beans.BeanUtils;

import com.google.cloud.ServiceOptions;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.Field;
import com.google.cloud.bigquery.FieldList;
import com.google.cloud.bigquery.FieldValue;
import com.google.cloud.bigquery.FieldValueList;
import com.google.cloud.bigquery.StandardSQLTypeName;
import com.google.cloud.bigquery.TableResult;
import com.google.common.collect.Maps;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * {@linkplain SqlDataSet}实体。
 * 
 * @author datagear@163.com
 */
@Slf4j
public class SqlDataSetEntity extends SqlDataSet implements DataSetEntity, CloneableEntity {
	private static final long serialVersionUID = 1L;

	/** 创建用户 */
	private User createUser;

	/** 创建时间 */
	private Date createTime;

	/** 权限 */
	private int dataPermission = PERMISSION_NOT_LOADED;

	private AnalysisProject analysisProject = null;

	public SqlDataSetEntity() {
		super();
		this.createTime = new Date();
	}

	public SqlDataSetEntity(SqlDataSet ds) {
		super();
		try {
			BeanUtils.copyProperties(ds, this);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public SqlDataSetEntity(String id, String name, List<DataSetProperty> properties, SchemaConnectionFactory connectionFactory, String sql,
			User createUser) {
		super(id, name, properties, connectionFactory, sql);
		this.createTime = new Date();
		this.createUser = createUser;
	}

	@Override
	public SchemaConnectionFactory getConnectionFactory() {
		return (SchemaConnectionFactory) super.getConnectionFactory();
	}

	@Override
	public void setConnectionFactory(ConnectionFactory connectionFactory) {
		if (connectionFactory != null && !(connectionFactory instanceof SchemaConnectionFactory))
			throw new IllegalArgumentException();

		super.setConnectionFactory(connectionFactory);
	}

	/**
	 * 获取{@linkplain SchemaConnectionFactory}。
	 * <p>
	 * 注意：
	 * </p>
	 * <p>
	 * 此方法与{@linkplain #getConnectionFactory()}功能一致，另参考{@linkplain #setShmConFactory(SchemaConnectionFactory)}。
	 * </p>
	 * 
	 * @return
	 */
	public SchemaConnectionFactory getShmConFactory() {
		return getConnectionFactory();
	}

	/**
	 * 设置{@linkplain SchemaConnectionFactory}。
	 * <p>
	 * 注意：
	 * </p>
	 * <p>
	 * 此方法与{@linkplain #setConnectionFactory(ConnectionFactory)}功能一致，主要用于序列化、反序列化此类实体对象（页面输入转换、ORM映射）时明确类型。
	 * </p>
	 * <p>
	 * 此方法名不应过长，某些数据库对标识符长度有限制，过长可能导致底层ORM的SQL语法错误（比如Oracle-12.2及以下版本限定标识符最长30个字符）。
	 * </p>
	 * 
	 * @param schemaConnectionFactory
	 */
	public void setShmConFactory(SchemaConnectionFactory schemaConnectionFactory) {
		setConnectionFactory(schemaConnectionFactory);
	}

	@Override
	public String getDataSetType() {
		return DataSetEntity.DATA_SET_TYPE_SQL;
	}

	@Override
	public void setDataSetType(String dataSetType) {
		// XXX 什么也不做，不采用抛出异常的方式，便于底层ORM统一SQL查询语句
		// throw new UnsupportedOperationException();
	}

	protected BigQuery getBigQuery(Schema schema) {
		// 测试bigQuery
		// -Djava.net.useSystemProxies=true -Dhttps.proxyHost=127.0.0.1
		// -Dhttps.proxyPort=1080
		if (StrUtil.isNotEmpty(schema.getUser())) {
			System.setProperty("java.net.useSystemProxies", "true");
			System.setProperty("https.proxyHost", schema.getUser());
			System.setProperty("https.proxyPort", schema.getPassword());
		}

		BigQueryClient.setEnv(ServiceOptions.CREDENTIAL_ENV_NAME, schema.getUrl());

		// 创建BigQuery客户端对象
		return BigQueryOptions.getDefaultInstance().getService();
	}

	@Override
	protected TemplateResolvedDataSetResult resolveResult(DataSetQuery query, List<DataSetProperty> properties, boolean resolveProperties)
			throws DataSetException {
		Schema schema = getConnectionFactory().getSchema();

		if (schema.isBigQuery()) {

			String sql = resolveSqlAsTemplate(getSql(), query);

			TemplateResolvedDataSetResult dataSetResult = new TemplateResolvedDataSetResult();
			// result.templateResult ==> sql
			dataSetResult.setTemplateResult(sql);

			try {
				TableResult rs = BigQueryClient.doQuery(getConnectionFactory().getSchema(), sql);
				// properties ==> meta
				ResolvedDataSetResult result = resolveResult(rs, query, properties, resolveProperties);

				dataSetResult = new TemplateResolvedDataSetResult(result.getResult(), result.getProperties(), sql);

			} catch (Throwable t) {
				log.warn("SQL:" + sql, t);
				throw new DataSetException(t);
			}
			return dataSetResult;
		}
		return super.resolveResult(query, properties, resolveProperties);
	}

	/**
	 * 解析结果。
	 * 
	 * @param cn
	 * @param result
	 * @param query
	 * @param properties
	 *        允许为{@code null}
	 * @param resolveProperties
	 * @return
	 * @throws Throwable
	 */
	protected ResolvedDataSetResult resolveResult(TableResult result, DataSetQuery query, List<DataSetProperty> properties, boolean resolveProperties)
			throws Throwable {
		List<DataSetProperty> bqProperties = new ArrayList<>();
		FieldList fields = result.getSchema().getFields();
		for (Field f : fields) {
			// TODO 转换名称 label
			DataSetProperty property = new DataSetProperty(f.getName(), toPropertyDataType(f.getType().getStandardType()));
			bqProperties.add(property);
		}
		// dataSetResult.setProperties(bqProperties);

		// // result.data ==> data
		List<Map<String, Object>> rawData = new ArrayList<>();

		DataSetPropertyValueConverter converter = createDataSetPropertyValueConverter();
		for (FieldValueList row : result.getValues()) {
			Map<String, Object> r = Maps.newHashMap();
			for (Field f : fields) {
				FieldValue c = row.get(f.getName());
				Object v = null;
				if (StandardSQLTypeName.DATE == f.getType().getStandardType() || StandardSQLTypeName.DATETIME == f.getType().getStandardType()
						|| StandardSQLTypeName.TIMESTAMP == f.getType().getStandardType()
						|| StandardSQLTypeName.TIME == f.getType().getStandardType()) {
					v = new Date(c.getTimestampValue() / 1000);
				} else if (StandardSQLTypeName.STRUCT == f.getType().getStandardType()) {
					FieldValueList map = c.getRecordValue();
					Map<String, Object> m = Maps.newLinkedHashMap();
					for (FieldValue fv : map) {
						List<FieldValue> value = fv.getRepeatedValue();
						if (value != null && value.size() == 2 && value.get(0) != null && value.get(1) != null) {
							if (value.get(0).getValue() == null || value.get(1).getValue() == null) {
								log.warn("{} : {}", f.getName(), JSONUtil.toJsonStr(value));
							}
							m.put(value.get(0).getStringValue(), value.get(1).getValue());
						} else {
							log.warn("{} : {}", f.getName(), JSONUtil.toJsonStr(value));
						}
					}
					v = JSONUtil.toJsonStr(m);
				} else if (StandardSQLTypeName.ARRAY == f.getType().getStandardType()) {
					v = JSONUtil.toJsonStr(c.getRepeatedValue());
				} else {
					v = c.getValue();
				}
				v = converter.convert(v, toPropertyDataType(f.getType().getStandardType()));
				r.put(f.getName(), v);
			}
			rawData.add(r);
		}

		return resolveResult(rawData, properties, query.getResultDataFormat());
	}

	/**
	 * 由BigQuery类型转换为{@linkplain DataSetProperty#getType()}。
	 * 
	 * @param sqlType
	 * @param columnName
	 *        允许为{@code null}，列名称
	 * @return
	 * @throws SQLException
	 * @throws SqlDataSetUnsupportedSqlTypeException
	 */
	protected String toPropertyDataType(StandardSQLTypeName sqlType) {
		switch (sqlType) {
		// 确定不支持的类型
		case BOOL:
			return DataType.BOOLEAN;
		case INT64:
		case NUMERIC:
			return DataType.NUMBER;
		case FLOAT64:
			return DataType.DECIMAL;
		case STRING:
			return DataType.STRING;
		case BYTES:
		case STRUCT:
		case ARRAY:
		case GEOGRAPHY:
			return DataType.UNKNOWN;
		case TIMESTAMP:
		case DATETIME:
			return DataType.TIMESTAMP;
		case DATE:
			return DataType.DATE;
		case TIME:
			return DataType.TIME;
		// return DataType.resolveDataType(value);
		default:
			return DataType.UNKNOWN;
		}
	}

	@Override
	public User getCreateUser() {
		return createUser;
	}

	@Override
	public void setCreateUser(User createUser) {
		this.createUser = createUser;
	}

	@Override
	public Date getCreateTime() {
		return createTime;
	}

	@Override
	public void setCreateTime(Date createTime) {
		this.createTime = createTime;
	}

	@Override
	public int getDataPermission() {
		return dataPermission;
	}

	@Override
	public void setDataPermission(int dataPermission) {
		this.dataPermission = dataPermission;
	}

	@Override
	public AnalysisProject getAnalysisProject() {
		return analysisProject;
	}

	@Override
	public void setAnalysisProject(AnalysisProject analysisProject) {
		this.analysisProject = analysisProject;
	}

	@Override
	public SqlDataSetEntity clone() {
		SqlDataSetEntity entity = new SqlDataSetEntity();
		BeanUtils.copyProperties(this, entity);

		return entity;
	}
}
